package com.swak.reactivex.threads;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.swak.metrics.MetricsFactory;
import com.swak.metrics.PoolMetrics;

/**
 * 指标执行功能的 Content
 * 
 * @author DELL
 */
@SuppressWarnings({ "rawtypes", "unchecked" })
public class MetricsContext implements ExecutorService, Context {

	private volatile PoolMetrics metrics;
	final ExecutorService executor;
	final String name;
	final int threads;

	public MetricsContext(String name, int threads, ExecutorService executor) {
		this.name = name;
		this.threads = threads;
		this.executor = executor;
	}

	@Override
	public void execute(Runnable command) {
		Object metric = metrics != null ? metrics.submitted() : null;
		this.executor.execute(new MetricsRunnable(metric, command));
	}

	@Override
	public void applyMetrics(MetricsFactory metricsFactory) {
		this.metrics = metricsFactory.createPoolMetrics(name, threads);
	}

	@Override
	public PoolMetrics<Object> getMetrics() {
		return this.metrics;
	}

	/**
	 * 可监控的
	 *
	 * @author lifeng
	 */
	class MetricsRunnable implements Runnable {
		private final Runnable command;
		private Object metric;

		public MetricsRunnable(Object metric, Runnable command) {
			this.command = command;
			this.metric = metric;
		}

		/**
		 * 正常执行代码
		 */
		@Override
		public void run() {
			Object usageMetric = null;
			if (metrics != null) {
				usageMetric = metrics.begin(metric);
			}
			boolean succeeded = true;
			try {
				command.run();
			} catch (Exception e) {
				succeeded = false;
			}
			if (metrics != null) {
				metrics.end(usageMetric, succeeded);
			}
		}
	}

	@Override
	public void shutdown() {
		executor.shutdown();
	}

	@Override
	public List<Runnable> shutdownNow() {
		return executor.shutdownNow();
	}

	@Override
	public boolean isShutdown() {
		return executor.isShutdown();
	}

	@Override
	public boolean isTerminated() {
		return executor.isTerminated();
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		return executor.awaitTermination(timeout, unit);
	}

	@Override
	public <T> Future<T> submit(Callable<T> task) {
		return executor.submit(task);
	}

	@Override
	public <T> Future<T> submit(Runnable task, T result) {
		return executor.submit(task, result);
	}

	@Override
	public Future<?> submit(Runnable task) {
		return executor.submit(task);
	}

	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
		return executor.invokeAll(tasks);
	}

	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException {
		return executor.invokeAll(tasks, timeout, unit);
	}

	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
		return executor.invokeAny(tasks);
	}

	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException, ExecutionException, TimeoutException {
		return executor.invokeAny(tasks, timeout, unit);
	}
}
